package com.yc.rpc.core.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yc.rpc.core.Metadata;
import com.yc.rpc.core.YcRpcResponse;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.SynchronousQueue;

@Component
@ChannelHandler.Sharable
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    private ConcurrentHashMap<String, SynchronousQueue> queueMap = new ConcurrentHashMap();

    private static Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);

    /**
     * 本方法用于接收服务端发送过来的消息
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        ByteBuf result = (ByteBuf) msg;
        byte[] result1 = new byte[result.readableBytes()];
        result.readBytes(result1);

        YcRpcResponse ycRpcResponse;
        try {
            ycRpcResponse = JSON.parseObject(new String(result1), YcRpcResponse.class);
        } catch (Exception e) {
            logger.error("发生异常了>>");
            return;
        }

        SynchronousQueue synchronousQueue = queueMap.get(ycRpcResponse.getRequestId());
        synchronousQueue.put(ycRpcResponse.getData());


        logger.info("服务端返回消息:[{}]", new String(result1));
        result.release();
    }

    /**
     * 本方法用于处理异常
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 当出现异常就关闭连接
        cause.printStackTrace();
//        ctx.close();
        logger.error("此处发生异常，连接关闭");

    }


    /**
     * 本方法用于向服务端发送信息
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        logger.info("已连接到RPC服务器.{}",ctx.channel().remoteAddress());
//
//        String msg = "{\"clazz\":\"com.rpc.yc.service.TestService\",\"method\":\"testRpc\",\"parameterTypes\":[\"java.lang.String\"],\"parameters\":[\"33\"]}\n";
//
//        ByteBuf encoded = ctx.alloc().buffer(4 * msg.length());
//        encoded.writeBytes(msg.getBytes());
//        ctx.write(encoded);
//        ctx.flush();
    }


    public SynchronousQueue<String> send(Channel channel, Metadata msg) {
        if (channel == null) {
            System.out.println("[send][连接不存在]");
        }
        if (!channel.isActive()) {
            System.out.printf("[send][连接(%s)未激活]", channel.id());
        }
        // 发送消息
        channel.writeAndFlush(Unpooled.copiedBuffer((JSONObject.toJSONString(msg)+ "\r\n").getBytes()));
        SynchronousQueue synchronousQueue = new SynchronousQueue();
        queueMap.put(msg.getRequestId(), synchronousQueue);
        return synchronousQueue;

    }
}
